package io.openmessaging.storage;

import io.openmessaging.Conf;
import io.openmessaging.Message;
import io.openmessaging.Topic;
import io.openmessaging.io.FileChannelWrapper;
import io.openmessaging.map.TopicQueueOffsetMap;
import io.openmessaging.utils.Assert;
import io.openmessaging.utils.Log;
import io.openmessaging.utils.Statistic;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author cuijd    2021/9/26 9:11
 */
public class SSDStorageNoQueue implements Storage {
    static class State {
        Message m;
        volatile boolean notified = false;
    }

    private static final class SSDFile {
        FileChannelWrapper channel;
        ConcurrentLinkedQueue<State> waitState = new ConcurrentLinkedQueue<>();
    }

    HashMap<Thread, Integer> threadId = new HashMap<>();
    AtomicInteger threadIdSeq = new AtomicInteger();
    SSDFile[] files = new SSDFile[4];
    private TopicQueueOffsetMap<FileIndex> indices = new TopicQueueOffsetMap<>();

    public SSDStorageNoQueue() {
        for (int i = 0; i < files.length; i++) {
            files[i] = new SSDFile();
            files[i].channel = new FileChannelWrapper(Conf.ssdPath + i);
            //根据已有数据恢复索引
            long readPos = 0;
            ByteBuffer head = ByteBuffer.allocate(9);//消息头，储存 topicId(1) queueId(2) offset(4) dataLength(2) 共9字节
            while (readPos < files[i].channel.size()) {
                head.clear();
                readPos += files[i].channel.readAndFlip(head, readPos);
                FileIndex index = indices.set(head.get(), head.getShort(), head.getInt(), new FileIndex((byte) i, readPos - 9, head.getShort()));
                readPos += index.length;
            }
        }

    }


    @Override
    public void save(Message m) {

        int i = threadId.computeIfAbsent(Thread.currentThread(), t -> threadIdSeq.getAndAdd(1) % files.length);
        SSDFile file = files[i];
        ByteBuffer head = ByteBuffer.allocate(9);//消息头，储存 topicId(1) queueId(2) offset(4) dataLength(2) 共9字节
        head.clear();
        head.put(m.topic.id).putShort((short) m.queueId).putInt((int) m.offset).putShort((short) m.data.remaining()).flip();
        //写入ssd
        int dataSize = m.data.remaining();
        long position = file.channel.write(head, m.data);
//        Log.i("write ",m);
        //消息索引
        FileIndex index = new FileIndex((byte) i, position, (short) dataSize);
        indices.set(m.topic.id, m.queueId, m.offset, index);

        State state = new State();
        state.m = m;
        Assert.isTrue(file.waitState.offer(state));

        if (file.waitState.size() >= 10) {
            file.channel.force();
            for (State s = file.waitState.poll(); s != null; s = file.waitState.poll()) {
                synchronized (s) {
//                    Log.i("notify ",s.m);
                    s.notify();
                    s.notified = true;
                }
            }
        } else {

            synchronized (state) {
                if (state.notified) {
//                    Log.i("needNotWait", m);
                } else {
                    try {
//                        Log.i("wait ",m);
                        long start = System.currentTimeMillis();
                        state.wait(1000);
                        if (System.currentTimeMillis() - start > 500) {
                            Log.i("等待超时");
                            file.channel.force();
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        System.exit(2);
                    }
                }
            }
        }
    }

    @Override
    public ByteBuffer get(Topic topic, int queueId, long offset) {
        return get(topic, queueId, offset, null);
    }

    /**
     * 先判断有没有落盘，如果没有，返回null，否则先从缓存中取，如果缓存取不到，再从 ssd 读
     */
    public ByteBuffer get(Topic topic, int queueId, long offset, Storage cache) {
        FileIndex index = this.indices.get(topic.id, queueId, offset);
        if (index == null)
            return null;

        ByteBuffer readBuffer = null;
        if (cache != null)
            readBuffer = cache.get(topic, queueId, offset);

        if (readBuffer == null) {
            readBuffer = ByteBuffer.allocate(index.length);
            files[index.fileId].channel.readAndFlip(readBuffer, index.position + 9);//跳过header

            Statistic.SSD.readCount.add(1);
            Statistic.SSD.readSize.add(readBuffer.remaining());
        }
        return readBuffer;
    }

}
